package com.example.repository;

import com.example.model.Product;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.util.Bytes;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Repository;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Optional;

@Repository
public class ProductRepository {

    private final Connection hbaseConnection;
    private static final String TABLE_NAME = "products";
    private static final byte[] CF_INFO = Bytes.toBytes("info");
    private static final byte[] COL_NAME = Bytes.toBytes("name");
    private static final byte[] COL_DESCRIPTION = Bytes.toBytes("description");
    private static final byte[] COL_IMAGE_URL = Bytes.toBytes("imageUrl");
    private static final byte[] COL_CATEGORY = Bytes.toBytes("category");
    private static final byte[] COL_PRICE = Bytes.toBytes("price");

    @Autowired
    public ProductRepository(Connection hbaseConnection) {
        this.hbaseConnection = hbaseConnection;
    }

    public List<Product> findAll() throws IOException {
        List<Product> products = new ArrayList<>();
        try (Table table = hbaseConnection.getTable(TableName.valueOf(TABLE_NAME));
             ResultScanner scanner = table.getScanner(new Scan())) {
            for (Result result : scanner) {
                products.add(mapToProduct(result));
            }
        }
        return products;
    }

    public Optional<Product> findById(String id) throws IOException {
        try (Table table = hbaseConnection.getTable(TableName.valueOf(TABLE_NAME))) {
            Get get = new Get(Bytes.toBytes(id));
            Result result = table.get(get);
            if (result.isEmpty()) {
                return Optional.empty();
            }
            return Optional.of(mapToProduct(result));
        }
    }

    public Product save(Product product) throws IOException {
        try (Table table = hbaseConnection.getTable(TableName.valueOf(TABLE_NAME))) {
            // If ID is null or empty, generate one
            if (product.getId() == null || product.getId().isEmpty()) {
                long timestamp = System.currentTimeMillis();
                product.setId("prod_" + timestamp);
            }
            
            Put put = new Put(Bytes.toBytes(product.getId()));
            put.addColumn(CF_INFO, COL_NAME, Bytes.toBytes(product.getName()));
            put.addColumn(CF_INFO, COL_DESCRIPTION, Bytes.toBytes(product.getDescription()));
            put.addColumn(CF_INFO, COL_IMAGE_URL, Bytes.toBytes(product.getImageUrl()));
            put.addColumn(CF_INFO, COL_CATEGORY, Bytes.toBytes(product.getCategory()));
            put.addColumn(CF_INFO, COL_PRICE, Bytes.toBytes(product.getPrice()));
            
            table.put(put);
        }
        return product;
    }

    public void deleteById(String id) throws IOException {
        try (Table table = hbaseConnection.getTable(TableName.valueOf(TABLE_NAME))) {
            Delete delete = new Delete(Bytes.toBytes(id));
            table.delete(delete);
        }
    }

    private Product mapToProduct(Result result) {
        Product product = new Product();
        product.setId(Bytes.toString(result.getRow()));
        product.setName(Bytes.toString(result.getValue(CF_INFO, COL_NAME)));
        product.setDescription(Bytes.toString(result.getValue(CF_INFO, COL_DESCRIPTION)));
        product.setImageUrl(Bytes.toString(result.getValue(CF_INFO, COL_IMAGE_URL)));
        product.setCategory(Bytes.toString(result.getValue(CF_INFO, COL_CATEGORY)));
        byte[] priceBytes = result.getValue(CF_INFO, COL_PRICE);
        if (priceBytes != null) {
            product.setPrice(Bytes.toDouble(priceBytes));
        }
        return product;
    }
} 